package pres.man.interview.bigdata;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 读取一亿条数据
 *
 * @author MAN
 * @date 2020-12-04-15:50
 */
public class Read {
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        input();
        long end = System.currentTimeMillis();
        System.out.println("处理一亿条数据耗时" + (end - start) / 1000 + "秒");
    }

    private static List<String> input() {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 3, 50, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());
        CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
        ArrayList<String> result = new ArrayList<>(5000000);
        try (BufferedReader bufferedReader = new BufferedReader(new FileReader("D:\\data.txt"))) {
            int count = 1;
            do {
                result.add(bufferedReader.readLine());
                // 每隔500w行数据处理一次
                if (count == 5000000) {
                    pool.execute(()->{
                        ArrayList<String> temp = new ArrayList<>(result);
                        try {
                            cyclicBarrier.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (BrokenBarrierException e) {
                            e.printStackTrace();
                        }
                        // 处理
                        temp.parallelStream().forEach(System.out::println);
                        // 处理完后清空数据
                        temp.clear();
                    });
                    // count复原
                    count = 1;
                    cyclicBarrier.await();
                    result.clear();
                    cyclicBarrier.reset();
                }
                count++;
            } while (bufferedReader.read() != -1);
        } catch (Exception e) {

        }
        pool.shutdown();
        return result;
    }
}
